package com.speily.bigdata.kafka;

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.FailureCallback;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.SuccessCallback;
import org.springframework.web.context.ContextLoader;

/**
 * @Auther: SPL
 * @Date: 2019-07-22 13:57
 * @Description: -kafka工具类，封装KafkaTemplate
 */
public class KafkaSendMsgUtils {

    public static <K,T>void sendMessage(String topic, Integer partition, Long timestamp,  K key, T data) {
        KafkaTemplate<K, T> kafkaTemplate = (KafkaTemplate<K, T>) ContextLoader.getCurrentWebApplicationContext().getBean("kafkaTemplate");
        ListenableFuture<SendResult<K, T>> listenableFuture;
        if (kafkaTemplate.getDefaultTopic().equals(topic)) {
            listenableFuture =  kafkaTemplate.sendDefault(partition,timestamp,key,data);
        }else {
            listenableFuture = kafkaTemplate.send(topic, partition, timestamp, key, data);
        }
        //发送成功回调
        SuccessCallback<SendResult<K, T>> successCallback = result -> {
            //成功业务逻辑
            System.out.println("成功");
        };
        //发送失败回调
        FailureCallback failureCallback = ex -> {
            //失败业务逻辑
            throw new RuntimeException(ex);
        };
        listenableFuture.addCallback(successCallback, failureCallback);
    }
}
